package com.xzc.apitest.tabletest

import com.xzc.apitest.source.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

object TableTest1 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")

    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    //首先创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env)

    //基于流创建一张表
    val dataTable: Table = tableEnv.fromDataStream(dataStream)
    val dataTable1: Table = tableEnv.fromDataStream(dataStream,
      "id", "timestamp", "temperature")
    val dataTable2: Table = tableEnv.fromDataStream(dataStream,
      'timestamp as 'ts, 'id as 'myId, 'temperature)
    val dataTable3: Table = tableEnv.fromDataStream(dataStream,
      'myId, 'ts)

    //调用table api进行转换
    val resultTable = dataTable
      .select("id, temperature")
      .filter("id == 'sensor_1'")

    //直接用sql实现
    tableEnv.createTemporaryView("dataTable", dataTable)
    tableEnv.createTemporaryView("dataView1", dataStream)
    tableEnv.createTemporaryView("dataView2", dataStream,
      'id, 'temperature, 'timestamp as 'ts)
    val sql: String = "select id, temperature from dataTable where id = 'sensor_1'"
    val resultSqlTable = tableEnv.sqlQuery(sql)

    resultTable.toAppendStream[(String, Double)].print("result")
    resultSqlTable.toAppendStream[(String, Double)].print("result sql")

    env.execute("table api example")
  }

}
